// 集群支持的聊天服务器示例
// 解决分布式环境下的状态共享和消息路由问题

#include "core/common.h"
#include <redis++/redis++.h>  // Redis C++ 客户端
#include <json/json.h>        // JSON 处理

class ClusterChatServer {
private:
    int serverSocket;
    int port;
    std::string nodeId;  // 节点ID
    
    // Redis 连接
    std::shared_ptr<sw::redis::Redis> redis;
    std::shared_ptr<sw::redis::Subscriber> subscriber;
    
    // 本地客户端连接 (只存socket连接，状态存Redis)
    std::unordered_map<int, std::string> localConnections;  // socket -> username
    std::mutex connectionsMutex;
    
    std::atomic<bool> running;

public:
    ClusterChatServer(int p = DEFAULT_PORT) 
        : port(p), nodeId(generateNodeId()), running(false) {
        
        // 连接Redis
        redis = std::make_shared<sw::redis::Redis>("tcp://redis:6379");
        subscriber = std::make_shared<sw::redis::Subscriber>(redis->subscriber());
        
        // 订阅集群消息
        subscriber->subscribe("chat_messages");
        subscriber->subscribe("user_events");
    }
    
    bool start() {
        // 创建socket (与原版相同)
        serverSocket = socket(AF_INET, SOCK_STREAM, 0);
        if (serverSocket < 0) {
            printLog("ERROR: Failed to create socket");
            return false;
        }
        
        // 绑定端口
        sockaddr_in serverAddr;
        serverAddr.sin_family = AF_INET;
        serverAddr.sin_addr.s_addr = INADDR_ANY;
        serverAddr.sin_port = htons(port);
        
        if (bind(serverSocket, (sockaddr*)&serverAddr, sizeof(serverAddr)) < 0) {
            printLog("ERROR: Failed to bind socket to port " + std::to_string(port));
            return false;
        }
        
        if (listen(serverSocket, MAX_CLIENTS) < 0) {
            printLog("ERROR: Failed to listen on socket");
            return false;
        }
        
        // 注册节点到Redis
        registerNode();
        
        // 启动Redis消息监听线程
        std::thread redisThread(&ClusterChatServer::handleRedisMessages, this);
        redisThread.detach();
        
        running = true;
        printLog("Cluster chat server started on port " + std::to_string(port));
        printLog("Node ID: " + nodeId);
        
        return true;
    }
    
    void run() {
        while (running) {
            sockaddr_in clientAddr;
            socklen_t clientLen = sizeof(clientAddr);
            
            int clientSocket = accept(serverSocket, (sockaddr*)&clientAddr, &clientLen);
            if (clientSocket < 0) {
                if (running) {
                    printLog("ERROR: Failed to accept client connection");
                }
                continue;
            }
            
            std::string clientIP = inet_ntoa(clientAddr.sin_addr);
            printLog("New connection from " + clientIP);
            
            // 为每个客户端创建处理线程
            std::thread clientThread(&ClusterChatServer::handleClient, this, clientSocket, clientIP);
            clientThread.detach();
        }
    }
    
private:
    std::string generateNodeId() {
        // 生成唯一节点ID (可以使用hostname + port)
        char hostname[256];
        gethostname(hostname, sizeof(hostname));
        return std::string(hostname) + ":" + std::to_string(port);
    }
    
    void registerNode() {
        // 在Redis中注册当前节点
        Json::Value nodeInfo;
        nodeInfo["nodeId"] = nodeId;
        nodeInfo["port"] = port;
        nodeInfo["timestamp"] = time(nullptr);
        
        Json::StreamWriterBuilder builder;
        std::string nodeInfoStr = Json::writeString(builder, nodeInfo);
        
        redis->setex("node:" + nodeId, 30, nodeInfoStr);  // 30秒过期
        printLog("Node registered in Redis: " + nodeId);
    }
    
    void handleRedisMessages() {
        // 监听Redis发布的消息
        subscriber->consume([this](std::string channel, std::string message) {
            if (channel == "chat_messages") {
                handleClusterChatMessage(message);
            } else if (channel == "user_events") {
                handleClusterUserEvent(message);
            }
        });
    }
    
    void handleClusterChatMessage(const std::string& message) {
        // 处理来自其他节点的聊天消息
        Json::Value root;
        Json::CharReaderBuilder builder;
        std::istringstream iss(message);
        std::string errs;
        
        if (Json::parseFromStream(builder, iss, &root, &errs)) {
            std::string fromNode = root["fromNode"].asString();
            std::string username = root["username"].asString();
            std::string content = root["content"].asString();
            
            // 只转发给本地连接的用户 (避免重复)
            if (fromNode != nodeId) {
                Message chatMsg(MSG_CHAT, username, content);
                broadcastToLocalClients(chatMsg);
            }
        }
    }
    
    void handleClusterUserEvent(const std::string& message) {
        // 处理用户上线/下线事件
        Json::Value root;
        Json::CharReaderBuilder builder;
        std::istringstream iss(message);
        std::string errs;
        
        if (Json::parseFromStream(builder, iss, &root, &errs)) {
            std::string event = root["event"].asString();
            std::string username = root["username"].asString();
            std::string fromNode = root["fromNode"].asString();
            
            if (fromNode != nodeId) {
                if (event == "join") {
                    Message joinMsg(MSG_CHAT, "Server", username + " joined the chat");
                    broadcastToLocalClients(joinMsg);
                } else if (event == "leave") {
                    Message leaveMsg(MSG_CHAT, "Server", username + " left the chat");
                    broadcastToLocalClients(leaveMsg);
                }
            }
        }
    }
    
    void handleClient(int clientSocket, const std::string& clientIP) {
        char buffer[BUFFER_SIZE];
        std::string username;
        bool authenticated = false;
        
        while (running) {
            memset(buffer, 0, BUFFER_SIZE);
            int bytesReceived = recv(clientSocket, buffer, BUFFER_SIZE - 1, 0);
            
            if (bytesReceived <= 0) {
                break;
            }
            
            std::string data(buffer);
            Message msg = Message::deserialize(data);
            
            if (!authenticated && msg.type == MSG_LOGIN) {
                username = msg.username;
                if (authenticateUserGlobally(username)) {
                    authenticated = true;
                    addUserToCluster(clientSocket, username, clientIP);
                    
                    // 发送登录成功消息
                    Message loginSuccess(MSG_LOGIN, "Server", "Login successful");
                    sendToClient(clientSocket, loginSuccess);
                    
                    // 广播用户加入事件到整个集群
                    broadcastUserEvent("join", username);
                    
                    // 发送全局在线用户列表
                    sendGlobalUserList(clientSocket);
                    
                    printLog("User '" + username + "' logged in from " + clientIP);
                } else {
                    Message loginFailed(MSG_ERROR, "Server", "Username already taken");
                    sendToClient(clientSocket, loginFailed);
                }
            } else if (authenticated) {
                switch (msg.type) {
                    case MSG_CHAT:
                        if (!msg.content.empty()) {
                            // 广播消息到整个集群
                            broadcastChatMessageToCluster(username, msg.content);
                            printLog("Message from " + username + ": " + msg.content);
                        }
                        break;
                        
                    case MSG_USER_LIST:
                        sendGlobalUserList(clientSocket);
                        break;
                        
                    case MSG_LOGOUT:
                        printLog("User '" + username + "' logged out");
                        goto cleanup;
                        
                    default:
                        break;
                }
            }
        }
        
cleanup:
        if (authenticated) {
            removeUserFromCluster(clientSocket, username);
            broadcastUserEvent("leave", username);
        }
        
        close(clientSocket);
        printLog("Client " + clientIP + " disconnected");
    }
    
    bool authenticateUserGlobally(const std::string& username) {
        // 在Redis中检查用户名是否已被使用 (全局检查)
        if (username.empty() || username == "Server") {
            return false;
        }
        
        // 使用Redis的SETNX原子操作确保用户名唯一
        auto result = redis->setnx("user:" + username, nodeId);
        if (result) {
            // 设置过期时间，防止死锁
            redis->expire("user:" + username, 3600);  // 1小时
            return true;
        }
        return false;
    }
    
    void addUserToCluster(int socket, const std::string& username, const std::string& ip) {
        // 添加到本地连接映射
        {
            std::lock_guard<std::mutex> lock(connectionsMutex);
            localConnections[socket] = username;
        }
        
        // 在Redis中记录用户信息
        Json::Value userInfo;
        userInfo["username"] = username;
        userInfo["nodeId"] = nodeId;
        userInfo["ip"] = ip;
        userInfo["timestamp"] = time(nullptr);
        
        Json::StreamWriterBuilder builder;
        std::string userInfoStr = Json::writeString(builder, userInfo);
        
        redis->hset("online_users", username, userInfoStr);
    }
    
    void removeUserFromCluster(int socket, const std::string& username) {
        // 从本地连接移除
        {
            std::lock_guard<std::mutex> lock(connectionsMutex);
            localConnections.erase(socket);
        }
        
        // 从Redis移除
        redis->del("user:" + username);
        redis->hdel("online_users", username);
    }
    
    void broadcastChatMessageToCluster(const std::string& username, const std::string& content) {
        // 构建消息JSON
        Json::Value messageJson;
        messageJson["fromNode"] = nodeId;
        messageJson["username"] = username;
        messageJson["content"] = content;
        messageJson["timestamp"] = time(nullptr);
        
        Json::StreamWriterBuilder builder;
        std::string messageStr = Json::writeString(builder, messageJson);
        
        // 发布到Redis，所有节点都会收到
        redis->publish("chat_messages", messageStr);
    }
    
    void broadcastUserEvent(const std::string& event, const std::string& username) {
        Json::Value eventJson;
        eventJson["fromNode"] = nodeId;
        eventJson["event"] = event;
        eventJson["username"] = username;
        eventJson["timestamp"] = time(nullptr);
        
        Json::StreamWriterBuilder builder;
        std::string eventStr = Json::writeString(builder, eventJson);
        
        redis->publish("user_events", eventStr);
    }
    
    void broadcastToLocalClients(const Message& msg) {
        std::string data = msg.serialize();
        
        std::lock_guard<std::mutex> lock(connectionsMutex);
        for (const auto& pair : localConnections) {
            send(pair.first, data.c_str(), data.length(), 0);
        }
    }
    
    void sendToClient(int socket, const Message& msg) {
        std::string data = msg.serialize();
        send(socket, data.c_str(), data.length(), 0);
    }
    
    void sendGlobalUserList(int socket) {
        // 从Redis获取全局在线用户列表
        auto onlineUsers = redis->hgetall("online_users");
        
        std::string userList;
        for (const auto& pair : onlineUsers) {
            if (!userList.empty()) userList += ", ";
            userList += pair.first;  // username
        }
        
        Message listMsg(MSG_USER_LIST, "Server", "Online users: " + userList);
        sendToClient(socket, listMsg);
    }
}; 